-module(reduce_server).

-export([for_each_reducer/3, new_reducer_table/3, reduce_service/3]).
-export([reduce/4, send_to_reduce/2, reducer/2]).

for_each_reducer({Key,Value}, TableName, Reducer) ->
    case ets:lookup(TableName, Key) of
	[] ->
	    Reducer_pid = spawn(reduce_server, reducer, [Reducer, []]),
	    ets:insert(all_reducer_pid, {Reducer_pid, over}),
	    if
		is_integer(Value) ->
		    Valist = [Value];
		true ->
		    Valist = [-1]
	    end,
	    ets:insert(TableName, {Key, Reducer_pid, Valist}),
	    {ok};
	[{T_Key, T_Pid, T_Values}] ->
	    ets:insert(TableName, {T_Key, T_Pid, [Value|T_Values]}),
	    {ok}
    end.

new_reducer_table([], _TableName, _Reducer) ->
    {ok};
new_reducer_table(KVList, TableName, Reducer) ->
    [{Key, Value}|T] = KVList,
    for_each_reducer({Key, Value}, TableName, Reducer),
    new_reducer_table(T, TableName, Reducer).

reduce_service(NodeId, Reducer, Output) ->
    Table_name_string = lists:append("reducer_pid_table", integer_to_list(NodeId)),
    ets:new(finished_map_table, [set, named_table, public]),
    ets:new(all_reducer_pid, [set, named_table, public]),
    reduce(Reducer, Output, Table_name_string, 0).

reduce(Reducer, Output, Table_name_string, N) ->
    NameString = lists:append(Table_name_string, integer_to_list(N)),
    TableName = list_to_atom(NameString),
    ets:new(TableName, [set, named_table, public]),
    receive
	{over, PID} ->
	    ets:insert(finished_map_table, {PID, over}),
	    OverNodeNumber = length(ets:tab2list(finished_map_table)),
	    if
		OverNodeNumber >= 4 ->
		    terminate_all_child();
		true ->
		    reduce(Reducer, Output, Table_name_string, N + 1)
	    end;
	L ->
	    new_reducer_table(L, TableName, Reducer),
	    First = ets:first(TableName),
	    if
		First /= '$end_of_table' ->
		    [{Key, Pid, Values}] = ets:lookup(TableName, First),
		    Pid ! {Key, Values},
		    send_to_reduce(ets:next(TableName, First), TableName);
		true ->
		    ok
	    end,
	    ets:delete(TableName),
	    reduce(Reducer, Output, Table_name_string, N+1)
    end.

send_to_reduce('$end_of_table', _TableName) ->
    ok;
send_to_reduce(Index, TableName) ->
    [{Key, Pid, Values}] = ets:lookup(TableName, Index),
    Pid ! {Key, Values},
    send_to_reduce(ets:next(TableName, Index), TableName).
    
reducer(Reducer, _List) ->
    receive
	{Key, Values} ->
	    reducer(Reducer, Reducer(Key, Values));
	over ->
	    ok
    end.

terminate_all_child() ->
    List = ets:tab2list(all_reducer_pid),
    send_list(List).

send_list([]) ->
    ok;
send_list(List) ->
    [{PID, over}|T] = List,
    PID ! over,
    send_list(T).

